package com.zy;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;

public class ChangeLogToTable {
    public static void main(String[] args) throws Exception{

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env);
        DataStream<Row> source = env.fromElements(
                Row.ofKind(RowKind.INSERT,"商品1",12),
                Row.ofKind(RowKind.INSERT,"商品2",12),
                Row.ofKind(RowKind.UPDATE_BEFORE,"商品1",12),
                Row.ofKind(RowKind.UPDATE_AFTER,"商品1",66)
        );
        Table table = tabEnv.fromChangelogStream(source);
        tabEnv.createTemporaryView("source_table",table);
        tabEnv.executeSql("select f0 as pid,sum(f1) as `all` from source_table group by f0").print();
        env.execute();
    }
}
